package com.atguigu.sparkstreaming.demos

import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
/**
 * Created by Smexy on 2022/8/22
 *
 *  程序的入口和核心编程API
 *
 *                入口               API
 *  sparkcore   SparkContext         RDD
 *  sparksql    SparkSession      DataFrame,DataSet
 *  sparkstreaming StreamingContext  DStream
 *
 *  DStream: 离散化流。  本质上是一个流，由若干个离散的批组成的流。
 *          spark的世界观：  流是无限的批。
 *
 *          DStream由无限个批次组成。可以对DStream定义一套计算逻辑，这个逻辑可以对DStream产生的每个批次都进行运算。
 *
 * ---------------------------------------
 *
 * StreamingContext的构造：
 *    方式一:    providing a Spark master URL and an appName：
 *          def this(master: String,appName: String,batchDuration: Duration)
 *
 *    方式二：   from a org.apache.spark.SparkConf
 *          def this(conf: SparkConf, batchDuration: Duration)
 *
 *    方式三:   from an existing org.apache.spark.SparkContext
 *         def this(sparkContext: SparkContext, batchDuration: Duration)
 *
 *
 *   batchDuration: Duration, 指一个时间间隔，DStream会按照这个时间间隔将流离散化为批。
 *      如果 batchDuration = 1min, 这个数据流，每一分钟产生的数据，都会封装为一个批次(RDD)
 *
 *
 *  -----------------------------------------
 *  套路:
 *
 *    ①创建入口  StreamingContext
 *    ②从StreamingContext中获取计算的流 DStream
 *        读取文件中实时写入的数据:  streamingContext.textFileStream("监控的目录")
 *        读取某个TCP端口中实时发送的文本数据:  streamingContext.socketTextStream()
 *    ③调用 DStream 提供的方法计算
 *    ④得到最终的计算结果
 *    ⑤输出结果
 *    ⑥启动App
 *
 *
 *
 *
 */
object HelloWorld {

  def main(args: Array[String]): Unit = {

    //①创建入口  StreamingContext
    val streamingContext = new StreamingContext("local[*]", "wordcount", Seconds(5))


   /*
      读取kafka的数据，当前App是一个Consumer角色。 消费者线程的配置

      必须配置的:
          bootstrap.servers: 集群地址
          key.deserializer： key的反序列化器
          value.deserializer： value的反序列化器

      可选:
          group.id: 消费者组id
          auto.offset.reset:  消费的策略。只在第一次启动时，有效。之后，从上次提交的位置往后消费。
                    latest:     只在当前组从未消费过当前主题时，从当前主题的最后的位置消费。
                    earliest:   只在当前组从未消费过当前主题时，从当前主题的最早的位置消费。

         enable.auto.commit:  是否开启自动提交offsets。如果配置了，会自动将当前组消费当前topic的offsets提交到
                              kafka集群上的 _consumer_offsets主题中

    */
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "sz220409test",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )

    /*
        要消费的主题。
            此处允许一个流同时订阅多个主题！
            但是，一般情况下，一个流应该只处理一个主题。
     */
    val topics = Array("topicD")

    /*
        从kafka中获取流的固定套路

        createDirectStream[K, V]:  每一个ConsumerRecord的K-V的类型，取决于生产者所生产的K-V的类型

        Kafka的工作流程：
            producer 生成  "hello world"----> ProducerRecord(topic,key,value)
                   ProducerRecord(topicA,null,hello world)-----序列化后 -------broker -------Consumer -----反序列化 ------ConsumerRecord(topicA,null,hello world)


-----------------------
def createDirectStream[K, V](
      ssc: StreamingContext：    App的入口
      locationStrategy: LocationStrategy:  大部分情况，就传  LocationStrategies.PreferConsistent
                  LocationStrategy 指App运行的Executor所在的机器和 Kafka的Broker所在的机器的位置关系(是不是同一个机器，还是同一个机架，还是同一个机房，还是没关系)

      consumerStrategy: ConsumerStrategy[K, V]： 大部分情况，就传 ConsumerStrategies.Subscribe
                消费者策略:
                        kafka中有两种消费者策略:
                              独立消费者(用于补数，对数):     offset不存储在kafka集群，每次消费时，自己指定位置，从哪个位置开始消费。
                                                消费者组自己指定要消费的topic和partition即可。

                                                ConsumerStrategies.assign

                              非独立消费者(常用):   offset存储在kafka集群，每次消费时，读取上次集群中存储的offsets，从那个位置往后消费。
                                              哪个线程消费哪个partition由 kafka broker用分配策略决定。
                                                消费者组只需要指定 topic即可。

                                                ConsumerStrategies.Subscribe
    ): InputDStream[ConsumerRecord[K, V]]
     */
    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    /*
          ③调用 DStream 提供的方法计算

          把DStream当RDD就行了！
     */
    val ds1: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map(word => (word, 1))
      .reduceByKey(_ + _)


    //⑤输出结果
    ds1.print(1000)

    //⑥启动App
    streamingContext.start()

    //等待发送终止命令，不发送，阻塞在这行，一直等
    streamingContext.awaitTermination()

  }

}
